package com.grf.kafka.java.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Headers;

import java.nio.charset.StandardCharsets;
import java.util.Map;

/**
 * @author grf
 * @apiNote 拦截器
 * @since 2024/10/27
 */
public class MyProductInterceptor implements ProducerInterceptor {
    /**
     * 消息发送时触发
     * @param record the record from client or the record returned by the previous interceptor in the chain of interceptors.
     * @return
     */
    @Override
    public ProducerRecord onSend(ProducerRecord record) {
        System.out.println("producer Record : " + record.toString());
        Headers headers = record.headers();
        headers.add("test", "grf".getBytes(StandardCharsets.UTF_8));
        return record;
    }

    /**
     * 收到服务端响应时触发
     * @param metadata The metadata for the record that was sent (i.e. the partition and offset).
     *                 If an error occurred, metadata will contain only valid topic and maybe
     *                 partition. If partition is not given in ProducerRecord and an error occurs
     *                 before partition gets assigned, then partition will be set to RecordMetadata.NO_PARTITION.
     *                 The metadata may be null if the client passed null record to
     *                 {@link org.apache.kafka.clients.producer.KafkaProducer#send(ProducerRecord)}.
     * @param exception The exception thrown during processing of this record. Null if no error occurred.
     */
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        System.out.println("acknowledgement recordMetadata:"+metadata.toString());
    }

    /**
     * 连接关闭时触发
     */
    @Override
    public void close() {
        System.out.println("producer closed");
    }

    /**
     * 整理配置项
     * @param configs 配置信息
     */
    @Override
    public void configure(Map<String, ?> configs) {
        System.out.println("=====producer config start======");
        for (Map.Entry<String, ?> entry : configs.entrySet()) {
            System.out.println("entry.key:"+entry.getKey()+" === entry.value:"+entry.getValue());
        }
        System.out.println("=====producer config end======");
    }
}
